#include "ModuleIOPortManager.h"
#include "MxBase/Log/Log.h"

namespace {
    const uint32_t MAX_QUEUE_LENGTH = 1000;
}

ModuleIOPortManager& ModuleIOPortManager::GetInstance() {
        static ModuleIOPortManager moduleIOPortManager;
        return moduleIOPortManager;
}

APP_ERROR ModuleIOPortManager::SendData(const std::string &port, const std::shared_ptr<void> &data) {
    ModuleQueue outputQueue;
    if (queues_.find(port) == queues_.end()) {
        queues_[port] = std::make_shared<BlockingQueue<std::shared_ptr<void>>>(MAX_QUEUE_LENGTH);
    }
    outputQueue = queues_[port];
    if (outputQueue.get() == nullptr) {
        LogError << port << " queue is nullptr";
        return APP_ERR_COMM_INVALID_PARAM;
    }
    outputQueue->Push(data, true);
    return APP_ERR_OK;
}
APP_ERROR ModuleIOPortManager::GetData(const std::string &port, std::shared_ptr<void> &data) {
    ModuleQueue inputQueue;
    if (queues_.find(port) == queues_.end()) {
        queues_[port] = std::make_shared<BlockingQueue<std::shared_ptr<void>>>(MAX_QUEUE_LENGTH);
    }
    inputQueue = queues_[port];
    if (inputQueue.get() == nullptr) {
        LogError << port << " queue is nullptr";
        return APP_ERR_COMM_INVALID_PARAM;
    }
    if (inputQueue->GetSize() > 50) {
        LogWarn << "[" <<port << "] size is " << inputQueue->GetSize();
    }
    auto ret = inputQueue->Pop(data);
    if (ret == APP_ERR_QUEUE_STOPED) {
        LogDebug << port <<  " input queue Stopped";
    } else if (ret != APP_ERR_OK || data == nullptr) {
        LogError << "Fail to get data from input queue for " << port
                << ", ret=" << ret;
    }
    return ret;
}


uint32_t ModuleIOPortManager::GetSize(const std::string &port) {
    if (queues_.find(port) == queues_.end()) {
        queues_[port] = std::make_shared<BlockingQueue<std::shared_ptr<void>>>(MAX_QUEUE_LENGTH);
    }
    auto q = queues_[port];
    if (q.get() == nullptr) {
        LogError << port << " queue is nullptr";
        return APP_ERR_COMM_INVALID_PARAM;
    }
    return q->GetSize();
}

void ModuleIOPortManager::Clear() {
    for (auto &queue : queues_) {
        queue.second->Stop();
        queue.second->Clear();
    }
}